package com.yangji.rxjavatest.rx.flowable;

import com.yangji.rxjavatest.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.schedulers.Schedulers;

/**
 * http://www.jianshu.com/p/9b1304435564
 */
public class RxFlowable01 {

    private static final String TAG = "MainActivity";

    public RxFlowable01() {
        Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR);


        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                //注意这句代码 这句话暂时注释后onNext无法收到消息（同步抛出异常，异步线程不会抛出异常）代表下游需要获取多少数据
                //或者根据上游发送事件的数量来request就行了, 比如这里request(3)就可以了.
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);

            }

            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }

        };
        flowable
                .subscribeOn(Schedulers.io())//加入这两句Subscription.request可以注释 不会抛出异常 但是依然不能收到数据
                .observeOn(AndroidSchedulers.mainThread())//加入这两句Subscription.request可以注释 但是依然不能收到数据
                .subscribe(subscriber);
    }
}
